package com.rock.code.province.tj.busi.kafkaDemo;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class KafkaProudcer2 {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("retries", 3);
        props.put("linger.ms", 1);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        while (true) {
            String topicName = "mykafka";
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    topicName, System.currentTimeMillis() + "", "今天天气不错哟yoyo=======>");
            producer.send(record, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if (e != null)
                        System.out.println("the producer has a error:" + e.getMessage());
                    else {
                        String f = "----------->>>partition=%s,offset=%s";
                        System.out.println(String.format(f, metadata.partition(), metadata.offset()));
                    }
                }
            });
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
        // producer.close();
    }
}